package com.zh.freechat.domain.notice;


import com.zh.freechat.infrastructure.mq.SseMsgMQProducer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.IdGenerator;
import org.springframework.util.SimpleIdGenerator;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Service
public class SseEmitterServer {
    private static IdGenerator idGenerator = new SimpleIdGenerator();
    /**
     * 使用map对象，便于根据userId来获取对应的SseEmitter，或者放redis里面
     */
    private final Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();

    private final SseMsgMQProducer sseMsgMQProducer;

    public SseEmitterServer(SseMsgMQProducer sseMsgMQProducer) {
        this.sseMsgMQProducer = sseMsgMQProducer;
    }

    /**
     * 创建用户连接并返回 SseEmitter
     *
     * @param userId 用户ID
     * @return SseEmitter
     */
    public SseEmitter connect(String userId) {
        SseEmitter sseEmitter = get(userId);
        if (sseEmitter==null) {
            sseEmitter = createSseEmitter(userId);
            sseEmitterMap.put(userId, sseEmitter);
        }

        log.info("创建新的sse连接，当前用户：{}", userId);
        return sseEmitter;
    }

    private SseEmitter createSseEmitter(String userId) {
        // 设置超时时间，0表示不过期。默认30秒，超过时间未完成会抛出异常：AsyncRequestTimeoutException
        SseEmitter sseEmitter = new SseEmitter(0L);
        // 注册回调
        sseEmitter.onCompletion(() -> {
            log.info("结束连接：{}", userId);
            remove(userId);
        });
        sseEmitter.onError(throwable -> {
            log.info("连接异常：{}", userId);
            remove(userId);
        });
        sseEmitter.onTimeout(() -> {
            log.info("连接超时：{}", userId);
            remove(userId);
        });
        return sseEmitter;
    }

    /**
     * 给指定用户发送信息
     * 当用户在线时：SSE推送消息
     * @return 当没有异常时，返回true
     */
    public boolean sendMessage(NoticeMessage message) {
        String receiver = message.getReceiver();
        if (sseEmitterMap.containsKey(receiver)) {
            String randomId = String.valueOf(idGenerator.generateId().getLeastSignificantBits()) ;
            log.debug("sse sendmessage:  messsage={}",message.toString());
            try {
                // sseEmitterMap.get(userId).send(message, MediaType.APPLICATION_JSON);
                sseEmitterMap.get(receiver).send(SseEmitter.event().name(message.channelName()).id(randomId).data(message));
                return true;
            } catch (IOException ex) {
                log.error("用户[{}]推送异常:{}", receiver, ex.getMessage());
                return false;
            }
        } else {
            //sse消息广播到MQ中，陈旧的消息直接忽略
            if (!message.isStale()) {
                sseMsgMQProducer.sendMsg(message);
            }
            return true;
        }
    }

    /**
     * 群发消息
     */
    public void batchSendMessage(NoticeMessage message, List<String> toUsers) {
        toUsers.forEach(e -> {
            NoticeMessage clone = message.clone();
            clone.setReceiver(e);
            sendMessage(clone);
        });
    }

    /**
     * 群发所有人
     */
    public void allSendMessage(NoticeMessage message) {
        String randomId = String.valueOf(idGenerator.generateId().getLeastSignificantBits()) ;
        sseEmitterMap.forEach((userId, sseEmitter) -> {
            try {
                sseEmitter.send(SseEmitter.event().name(message.channelName()).id(randomId).data(message));
            } catch (IOException ex) {
                log.error("用户[{}]推送异常:{}", userId, ex.getMessage());
            }
        });
    }

    /**
     * 移除用户连接
     */
    private void remove(String userId) {
        sseEmitterMap.remove(userId);
        log.info("移除用户：{}", userId);
    }

    private SseEmitter get(String userId) {
        log.info("获取用户：{}", userId);
        return sseEmitterMap.get(userId);
    }

    /**
     * 获取当前连接信息
     */
    public  List<String> getIds() {
        return new ArrayList<>(sseEmitterMap.keySet());
    }

    /**
     * 获取当前连接数量
     */
    public  int getUserCount() {
        return sseEmitterMap.size();
    }

}